﻿#include <net/udpchannel.h>
#include <net/netreading.h>

#ifdef Q_OS_WIN
#include <winsock2.h>
#ifndef SIO_UDP_CONNRESET
#  ifndef IOC_VENDOR
#    define IOC_VENDOR 0x18000000
#  endif
#  ifndef _WSAIOW
#    define _WSAIOW(x,y) (IOC_IN|(x)|(y))
#  endif
#  define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR,12)
#endif //SIO_UDP_CONNECTRESET
#endif // Q_OS_WIN

QNET_USING_NAMESPACE

QNET_NAMESPACE_BEGIN
/**
 * @brief The InnerProc class 异步处理类
 */
class InnerProc :public TaskEntry
{
public:
    InnerProc(UdpChannel *owner, QByteArray &data, QHostAddress host, quint16 port);
    // TaskEntry interface
public:
    void onProc();
    void onDiscard();
private:
    UdpChannel *m_owner;
    QByteArray m_data;
    QHostAddress m_host;
    quint16 m_port;
};
QNET_NAMESPACE_END

UdpChannel::UdpChannel(QObject *parent):QObject (parent)
  ,m_readQueue(2000)
{
    m_udp = Q_NULLPTR;
    m_reading = Q_NULLPTR;
}

bool UdpChannel::bind(const QHostAddress &localAddress, quint16 localPort)
{
    bool  init  = false;
    if(!m_udp)
    {
        m_udp = new QUdpSocket(this);
        init= m_udp->bind(localAddress, localPort, QUdpSocket::ReuseAddressHint);
        if(!init)
        {
            m_udp->close();
            m_udp->deleteLater();
            m_udp = nullptr;
        }else
        {
#ifdef Q_OS_WIN
            // 初始化buffer大小，
            qintptr sck = m_udp->socketDescriptor();
            int recvBuf = 100*1024*1024;// 默认设置为100MB
            int ret =
                    setsockopt(SOCKET(sck), SOL_SOCKET, SO_RCVBUF, reinterpret_cast<const char*>(&recvBuf), sizeof(int));
            if(ret)
            {
                 qWarning()<<"socket rcv set error "<<ret;
            }
            int sendBuf = 100*1024*1024;// 默认设置为100MB
            ret = setsockopt(SOCKET(sck), SOL_SOCKET, SO_SNDBUF, reinterpret_cast<const char*>(&sendBuf), sizeof(int));
            if(ret)
            {
                qWarning()<<"socket snd set error" << ret;
            }

            DWORD dwBytesReturned = 0;
            int bNewBehavior = 0;
            // 取消UDP不可达消息通知，提升性能。
            if (::WSAIoctl(SOCKET(sck), SIO_UDP_CONNRESET, &bNewBehavior, sizeof(bNewBehavior),
                           Q_NULLPTR, 0, &dwBytesReturned, Q_NULLPTR, Q_NULLPTR) == SOCKET_ERROR) {
                qWarning() << "WSAIoctl  error: " << WSAGetLastError();
            }
#endif
//            connect(m_udp,SIGNAL(readyRead()), this, SLOT(read()));
            connect(m_udp, &QUdpSocket::readyRead, this, &UdpChannel::read);
            connect(this, &UdpChannel::doCloseSocket, this, &UdpChannel::onCloseSocket);
        }
    }
    return init;
}

void UdpChannel::setRemote(const QHostAddress &remoteAddress, quint16 remotePort)
{
    m_remoteAddress = remoteAddress;
    m_remotePort = remotePort;
}

void UdpChannel::setReading(NetReading *reading)
{
    m_reading = reading;
}

int UdpChannel::write(const QByteArray &data)
{
    qint64 len = m_udp->writeDatagram(
                data.constData(),
                data.size(),
                m_remoteAddress,
                m_remotePort);
    return int(len);
}

int UdpChannel::writeTo(const QByteArray &data, const QHostAddress &remoteAddr, quint16 remotePort)
{
    qint64 len = m_udp->writeDatagram(
                data.constData(),
                data.size(),
                remoteAddr,
                remotePort);
    return int(len);
}


void UdpChannel::close()
{
    emit doCloseSocket();
}

void UdpChannel::read()
{
    QHostAddress remoteHost;
    quint16 remotePort = 0;
    qint64 readLen = 0;
    while (m_udp->hasPendingDatagrams())
    {
        m_readDatagram.resize(static_cast<int>(m_udp->pendingDatagramSize()));
        readLen = m_udp->readDatagram(m_readDatagram.data(),
                                      m_readDatagram.size(),
                                      &remoteHost,
                                      &remotePort);
        if(readLen > 0)
        {
//            qWarning() << "Udp channel recv" << remoteHost << remotePort;
            m_remoteAddress = remoteHost;
            m_remotePort = remotePort;
            InnerProc *proc =  new InnerProc(this, m_readDatagram, remoteHost, remotePort);
            if(m_readQueue.add(proc) == false)
            {
                delete proc;
                qWarning() << "UdpDataOperator Read queue is fill!";
            }
        }
    }
}

void UdpChannel::onCloseSocket()
{
    if(m_udp)
    {
        m_udp->close();
    }
}

QHostAddress UdpChannel::remoteAddress() const
{
    return m_remoteAddress;
}

quint16 UdpChannel::localPort() const
{
    return m_udp->localPort();
}

QHostAddress UdpChannel::localAddress() const
{
    return m_udp->localAddress();
}

quint16 UdpChannel::remotePort() const
{
    return m_remotePort;
}

InnerProc::InnerProc(UdpChannel *owner, QByteArray &data, QHostAddress host, quint16 port)
{
    m_owner = owner;
    m_data = data;
    m_host = host;
    m_port = port;
}

void InnerProc::onProc()
{
    NetReading *reading = m_owner->m_reading;
    if(reading)
    {
        reading->setHost(m_host);
        reading->setPort(m_port);

        reading->onProc(m_data);
    }
}

void InnerProc::onDiscard()
{
    qWarning() << QObject::tr("Udp InnerProc  was canceled！");
}
